//
// Created by Administrator on 2025/8/16.
//

#include "aliyunclient.h"
#include "FreeRTOS.h"
#include "task.h"
#include "queue.h"
#include "lwip/netdb.h"
#include "lwip/sockets.h"
#include "printf_rtos.h"
#include "mqttAPI.h"
#include "MQTTPacket.h"
#include "MQTTConnect.h"
#include "cJSON.h"
#include "data_codec.h"
#include "event_groups.h"
#include "semphr.h"
#include "event_groups.h"
#include "freertos_demo.h"


#define ALIYUN_MQTT_SERVER_NAME                    "iot-06z00b3p62co4fb.mqtt.iothub.aliyuncs.com"
#define ALIYUN_MQTT_SERVER_PORT                    "1883"
#define ALIYUN_MQTT_SERVER_USER                    "alientekf407&k1mn8ececb3"
#define ALIYUN_MQTT_SERVER_PASS                    "6ee480c1a08f4eea7a9a8e1671fcfbce09a1dec7ef75686a220385c2b73a6999"
#define ALIYUN_MQTT_SERVER_CLIENTID                "k1mn8ececb3.alientekf407|securemode=2,signmethod=hmacsha256,timestamp=1755667083777|"
#define ALIYUN_MQTT_SERVER_PUB_TOPIC               "/sys/k1mn8ececb3/alientekf407/thing/event/property/post"
#define ALIYUN_MQTT_SERVER_SUB_TOPIC               "/sys/k1mn8ececb3/alientekf407/thing/service/property/set"
#define ALIYUN_MQTT_SERVER_SUB_QOS                 1
#define ALIYUN_MQTT_SERVER_KEEP_ALIVE_INTERVAL_S   60
#define LENGTH_SENDBUF              256
#define LENGTH_RECVBUF              256

int g_mqtt_aliyun_socket = -1;
volatile int g_mqtt_aliyun_socket_status=-1;
//温度传感器数值（℃）
extern float internal_temp_celsius;
//MQTT成功连接信号量
// extern SemaphoreHandle_t g_mqtt_server_connect_sem;
extern SemaphoreHandle_t g_dns_mutex;
extern EventGroupHandle_t g_init_eventgroup;
//socket状态枚举
typedef enum {
    ALIYUN_SERVER_DISCONNECTED,
    ALIYUN_SERVER_CONNECTED
}aliyun_server_socket_t;


//MQTT 客户端结构体
//status指的是此时socket的状态，0表示连接断开，1表示连接成功
typedef struct {
    int socket;
    volatile int status;
}aliyun_client_t;




/**
 * 从新大陆服务器socket接收数据
 * @param buf 接收数据的缓冲区指针
 * @param len 期望接收的数据长度
 * @return 实际接收的字节数，-1表示连接断开或错误，0表示无数据可读
 */
int aliyun_transport_read(unsigned char* buf, int len)
{
    // 检查 socket 句柄是否有效
    int bytes_read=0;
    if (g_mqtt_aliyun_socket < 0) {
        g_mqtt_aliyun_socket_status=ALIYUN_SERVER_DISCONNECTED;
        return -1;
    }

    // 从socket接收数据
    bytes_read=recv(g_mqtt_aliyun_socket, buf, len, 0);

    // 处理接收成功的数据
    if (bytes_read>0)
    {
        g_mqtt_aliyun_socket_status=ALIYUN_SERVER_CONNECTED;
        return bytes_read;
    }

    // 处理对端关闭连接的情况
    if (bytes_read==0)
    {
        printf_rtos("Transport: Peer closed connection.\n");
        g_mqtt_aliyun_socket_status=ALIYUN_SERVER_DISCONNECTED;
        return -1;
    }

    // 处理非阻塞模式下无数据可读的情况
    if (errno == EAGAIN)
    {
        return 0;
    }

    // 处理接收错误，连接丢失的情况
    printf_rtos("Transport: recv error, errno=%d. Connection lost.\n", errno);
    g_mqtt_aliyun_socket_status=ALIYUN_SERVER_DISCONNECTED;
    return -1;
}



/**
 * 向新大陆MQTT服务器发送数据
 *
 * @param buf 要发送的数据缓冲区指针
 * @param len 要发送的数据长度
 * @return 成功时返回发送的字节数，失败时返回-1
 *
 * 该函数通过已建立的socket连接向新大陆MQTT服务器发送指定长度的数据。
 * 在发送前会检查socket连接是否有效。
 */
int aliyun_transport_write(unsigned char* buf, int len)
{
    // 检查socket连接是否有效，无效则直接返回错误
    if (g_mqtt_aliyun_socket < 0) {
        return -1;
    }
    // 调用系统send函数发送数据
    return send(g_mqtt_aliyun_socket, buf, len, 0);
}




/**
 * @brief 获取下一个数据包ID
 *
 * 该函数用于生成递增的数据包ID，ID范围为1-65535，达到最大值后自动回绕到1
 * 使用静态变量保证每次调用都能获得唯一的递增ID
 *
 * @return uint16_t 返回下一个数据包ID，范围1-65535
 */
uint16_t get_next_packetid(void)
{
    static uint16_t packetid=0;

    // 当ID达到最大值时回绕到0，下次递增后变为1
    if (packetid >= 65535)
        packetid=0;

    // 递增并返回数据包ID
    return ++packetid;
}

/**
 * 释放JSON字符串内存
 * @param buf: 指向JSON字符串指针的指针，函数会释放该指针指向的内存
 * @return: 成功释放返回true，参数无效返回false
 */
bool pub_json_free(char **buf)
{
    //避免空指针/野指针异常
    if (!buf||!*buf)   return false;
    //释放JSON字符串内存
    cJSON_free(*buf);
    //避免野指针
    *buf=NULL;
    return true;
}


void aliyunClient_task(void *pvParms)
{

    xEventGroupWaitBits(g_init_eventgroup,BIT_NTP_READY,pdFALSE,\
        pdTRUE,portMAX_DELAY);
    UNUSED(pvParms);
    int ret=0;//返回值中间变量
    int len=0;//数据长度中间变量
    int dup=-1;//重发标志
    uint32_t reconnect_delay=500;//重连间隔
    uint16_t aliyun_packetid=0;//新大陆MQTT数据包ID
    TickType_t last_active_tick=xTaskGetTickCount();//最后一次活动时间
    TickType_t last_pingreq_tick=last_active_tick;//最后一次发送PINGREQ的时间
    bool is_wait_pingresp=false;//是否等待PINGRESP
    struct timeval mqttOT={0};//服务器 SOCKET超时配置结构体
    mqttOT.tv_sec=1;//配置为1S超时

    struct addrinfo hints={0};//DNS解析配置信息结构体
    hints.ai_family = AF_INET;//IPV4服务器
    hints.ai_socktype = SOCK_STREAM;//TCP连接
    struct addrinfo *res=NULL;//DNS解析结果结构体指针

    struct sockaddr_in server_addr;//服务器地址结构体
    char serverip4addr[16]={0};//解析后的IP地址缓冲区

    //收发缓冲区
    static uint8_t sendBuf[LENGTH_SENDBUF]={0};
    static uint8_t recvBuf[LENGTH_RECVBUF]={0};

    for (;;)
    {
        //DNS解析部分
        printf_rtos("aliyun task running...\n");
        printf_rtos("waiting for dns_mutex\n");
        xSemaphoreTake(g_dns_mutex,portMAX_DELAY);
        printf_rtos("DNS Resolving...\n");
        //获取DNS解析结果，返回的ret为0时成功。
        ret=getaddrinfo(ALIYUN_MQTT_SERVER_NAME,ALIYUN_MQTT_SERVER_PORT,&hints,&res);
        if (ret!=0)
        {
            printf_rtos("DNS resolution failed. Retrying...errno:%d.\n",ret);
            xSemaphoreGive(g_dns_mutex);
            vTaskDelay(1000);
            continue;
        }
        xSemaphoreGive(g_dns_mutex);
        server_addr=*(struct sockaddr_in *)res->ai_addr;//解析的得到的服务器地址结构体

        // 将服务器IPv4地址从网络字节序转换为点分十进制字符串格式并存储到serverip4addr缓冲区中
        inet_ntop(AF_INET, &server_addr.sin_addr, serverip4addr, sizeof(serverip4addr));
        printf_rtos("DNS resolution successful.IP:%s\n",serverip4addr);
        //创建socket连接
        printf_rtos("Socket Creating...\n");
        // 创建一个socket连接，参数说明如下：
        // res->ai_family: 地址族，AF_INET表示使用IPv4
        // res->ai_socktype: 套接字类型，SOCK_STREAM表示TCP连接
        // res->ai_protocol: 协议类型，通常为0，表示使用默认协议
        g_mqtt_aliyun_socket=socket(res->ai_family,res->ai_socktype,res->ai_protocol);
        //socket创建失败了，可能是端口被占用，释放掉并重新尝试
        if (g_mqtt_aliyun_socket<0)
        {
            printf_rtos("Socket create failed.Retrying...\n");
            freeaddrinfo(res);
            res=NULL;
            goto reconnect;
        }
        //创建完成后DNS已经没用了，服务器地址信息已经保存到server_addr结构体中
        //及时释放，内存安全
        freeaddrinfo(res);
        printf_rtos("Socket create successfully\n");
        // 设置socket接收超时时间，mqttOT.tv_sec=1表示接收数据时最多等待1秒
        setsockopt(g_mqtt_aliyun_socket, SOL_SOCKET, SO_RCVTIMEO, &mqttOT, sizeof(mqttOT));
        // 修改mqttOT结构体中的超时时间为10秒
        mqttOT.tv_sec = 10;
        // 设置socket连接超时时间，SO_CONTIMEO选项用于设置连接操作的超时时间
        setsockopt(g_mqtt_aliyun_socket, SOL_SOCKET, SO_CONTIMEO, &mqttOT, sizeof(mqttOT));
        printf_rtos("Connecting to %s:%s\n",serverip4addr,ALIYUN_MQTT_SERVER_PORT);
        //连接服务器
        ret=connect(g_mqtt_aliyun_socket,(struct sockaddr *)&server_addr,sizeof(server_addr));
        //连接失败或超时，释放socket重连
        if (ret<0)
        {
            printf_rtos("TCP Connecting Aliyun failed\n");
            //释放socket
            closesocket(g_mqtt_aliyun_socket);
            goto reconnect;
        }
        printf("TCP Connected.IP:%s:%s\n",serverip4addr,ALIYUN_MQTT_SERVER_PORT);
        //TCP连接成功，配置MQTT协议连接参数
        //MQTT 连接参数结构体
        MQTTPacket_connectData mqtt_conndata = MQTTPacket_connectData_initializer;
        mqtt_conndata.clientID.cstring=ALIYUN_MQTT_SERVER_CLIENTID;
        mqtt_conndata.keepAliveInterval=ALIYUN_MQTT_SERVER_KEEP_ALIVE_INTERVAL_S;
        mqtt_conndata.username.cstring=ALIYUN_MQTT_SERVER_USER;
        mqtt_conndata.password.cstring=ALIYUN_MQTT_SERVER_PASS;
        //根据结构体内容生成对应的MQTT协议的TCP数据包
        len = MQTTSerialize_connect(sendBuf,LENGTH_SENDBUF,&mqtt_conndata);
        //将连接信息发送到对端服务器
        ret=aliyun_transport_write(sendBuf,len);
        //发送失败，重连
        if (ret<0)
            goto reconnect;

        //发送成功后，服务器会返回连接成功标识CONNACK，为数据包第一个字节的高四位
        ret=MQTTPacket_read(recvBuf,LENGTH_RECVBUF,aliyun_transport_read);
        //接收到的非CONNACK标识，重连
        if (ret!=CONNACK)
        {

            printf_rtos("Can't recv CONNACK from server.\n");
            closesocket(g_mqtt_aliyun_socket);
            res=NULL;
            goto reconnect;
        }
        //连接成功，记录为一次有效活动
        last_active_tick=xTaskGetTickCount();
        printf_rtos("MQTT connect to ALIYUN successfully.\n");

        //开始订阅服务器的Publish topic
        printf_rtos("Prepare to subscribe to Topic.\n");
        //MQTT专属字符串结构体MQTTString.MQTTString_initializer为宏，初始化此结构体
        MQTTString sub_topic = MQTTString_initializer;

        //初始化是否重发，此处为否
        if (dup==-1) dup=0;
        // 设置订阅的QoS（Quality of Service服务质量）等级，QoS等级决定了消息传递的可靠性
        // QoS 0：最多一次传递，消息可能丢失
        // QoS 1：至少一次传递，消息不会丢失但可能重复
        // QoS 2：恰好一次传递，最高可靠性但开销最大
        int qos = ALIYUN_MQTT_SERVER_SUB_QOS;  // 当前使用QoS等级1
        // 获取下一个MQTT数据包ID，用于标识本次订阅请求
        // MQTT协议要求packetid为1-65535之间的唯一标识符，0为保留值
        aliyun_packetid = get_next_packetid();
        sub_topic.cstring=ALIYUN_MQTT_SERVER_SUB_TOPIC;//配置订阅的Topic
        //生成及发送订阅数据包
        len=MQTTSerialize_subscribe(sendBuf,LENGTH_SENDBUF,dup,aliyun_packetid,1,&sub_topic,&qos);
        ret=aliyun_transport_write(sendBuf,len);
        //发送失败或超时，重试
        if (ret<0)
            goto reconnect;
        //订阅若成功服务器会返回SUBACK
        ret=MQTTPacket_read(recvBuf,LENGTH_RECVBUF,aliyun_transport_read);
        if (ret!=SUBACK)
        {
            closesocket(g_mqtt_aliyun_socket);
            printf_rtos("Can't recv SUBACK from server.\n");
            //配置为重发订阅
            dup=1;
            goto reconnect;
        }
        //接收到SUBACK处理逻辑

        uint16_t ack_packetid=0;
        int grantedQoS_cnt=0;
        int grantedQoS[10]={0};
        // 调用MQTT库函数反序列化SUBACK包，检查订阅确认信息
        // 参数说明:
        // - &ack_packetid: 用于存储服务器返回的packetid，应与发送的订阅packetid一致
        // - 10: grantedQoS数组的最大长度，表示最多可以处理10个QoS返回值
        // - &grantedQoS_cnt: 用于存储实际返回的QoS数量
        // - grantedQoS: 存储服务器返回的每个主题的QoS等级数组
        // - recvBuf: 接收到的SUBACK数据包缓冲区
        // - LENGTH_RECVBUF: 接收缓冲区的长度
        if (MQTTDeserialize_suback(&ack_packetid, 10, &grantedQoS_cnt, grantedQoS, recvBuf, LENGTH_RECVBUF))
        {
            // 检查服务器返回的packetid是否与我们发送的订阅packetid一致，并且确认QoS等级不是失败标识(0x80)
            if (ack_packetid == aliyun_packetid && grantedQoS[0] != 0x80)
                printf_rtos("Subscription successful, current Qos:%d\n",grantedQoS[0]);
        }
        else
        {
            //订阅失败 重连
            printf_rtos("Subscription failed or refused\n");
            goto reconnect;
        }
        //订阅成功了 算是一次有效活动
        last_active_tick=xTaskGetTickCount();
        //重置重复标志
        dup=-1;
        printf_rtos("Sub topic:%s successfully.\n",sub_topic.cstring);
        //重置重连间隔
        reconnect_delay = 500;

        //释放MQTT连接成功信号量
        // xSemaphoreGive(g_mqtt_server_connect_sem);
        xEventGroupSetBits(g_init_eventgroup,BIT_MQTT_READY);
        xEventGroupWaitBits(g_init_eventgroup,BIT_CODEC_READY,pdFALSE,pdTRUE,portMAX_DELAY);
        xEventGroupWaitBits(g_init_eventgroup,BIT_DECODEC_READY,pdFALSE,pdTRUE,portMAX_DELAY);
        //内层循环，MQTT主进程
        for (;;)
        {
            //读接收数据
            ret=MQTTPacket_read(recvBuf,LENGTH_RECVBUF,aliyun_transport_read);
            if (ret>0)//接收到数据
            {
                if (ret==PUBLISH)//如果是发布的数据类型
                {
                    uint8_t dup=0;
                    int qos=-1;
                    uint8_t retained=0;
                    uint16_t packetid=0;
                    MQTTString topicName=MQTTString_initializer;
                    uint8_t *payload=NULL;
                    int payloadlen=0;
                    // 解包接收到的MQTT PUBLISH数据包
                    // 参数说明:
                    // - dup: 指向存储重复发送标志的指针，1表示是重复发送的数据包
                    // - qos: 指向存储服务质量等级的指针，值为0、1或2
                    // - retained: 指向存储保留标志的指针，1表示这是保留消息
                    // - packetid: 指向存储数据包ID的指针，用于QoS>0时的消息确认
                    // - topicName: 指向存储主题名称的MQTTString结构体指针
                    // - payload: 指向存储消息载荷数据的指针
                    // - payloadlen: 指向存储载荷数据长度的指针
                    // - recvBuf: 接收到的原始数据缓冲区
                    // - LENGTH_RECVBUF: 接收缓冲区的总长度
                    ret=MQTTDeserialize_publish(&dup,&qos,&retained,&packetid,&topicName,&payload,&payloadlen,recvBuf,LENGTH_RECVBUF);
                    //解包成功
                    if (ret==1)
                    {
                        //如果Qos为1，则客户端需要回复一个确认包
                        if (qos==1)
                        {
                            /*
                            * 序列化PUBACK确认包并发送
                            * 如果发送失败则跳转到重连流程
                            */
                            len=MQTTSerialize_puback(sendBuf,LENGTH_SENDBUF,packetid);
                            int puback_ret=aliyun_transport_write(sendBuf,len);
                            if (puback_ret<0)
                                goto reconnect;
                        }
                        printf_rtos("<<< Recv PUBLISH on topic '%.*s': %.*s\n",
                        topicName.lenstring.len, topicName.lenstring.data,
                        payloadlen, payload);

                        //创建一个传感器数据结构体，将接收到的数据复制到结构体的payload字段中
                        sensor_raw_data_t temp_sensor_data={NULL,0};
                        temp_sensor_data.payload=pvPortMalloc(sizeof(uint8_t)*payloadlen);//开辟空间存放数据
                        //分配内存失败
                        if (temp_sensor_data.payload==NULL)
                        {
                            printf_rtos("Error: Failed to allocate memory for payload!\n");
                            continue; // 放弃这条消息，继续外层 for 循环
                        }
                        //分配内存成功，复制数据
                        temp_sensor_data.payload_len=payloadlen;
                        memcpy(temp_sensor_data.payload,payload,payloadlen);
                        //发送到队列，等待解码任务处理
                        ret=xQueueSend(sensor_recv_queue,&temp_sensor_data,0);
                        //发送失败了，需要释放掉开辟的内存避免内存泄露
                        if (ret!=pdPASS)
                        {
                            printf_rtos("Error: Sensor receive queue is full. Dropping message.\n");
                            vPortFree(temp_sensor_data.payload);
                        }
                    }
                    //一次接收到的数据包处理完毕，更新最后一次有效活动时间
                    last_active_tick=xTaskGetTickCount();
                }
                else if (ret== PINGRESP) // 如果是心跳回复
                {
                    printf_rtos("--- Ping Response received ---\n");
                    //接收到对端的心跳回应了，置等待心跳回应标志位为假
                    is_wait_pingresp=false;
                    last_active_tick=xTaskGetTickCount();
                }
                //接收结束，清空缓冲区
                memset(recvBuf,'\0',LENGTH_RECVBUF);
            }
            else   //没接收到数据 重连
            {
                if (g_mqtt_aliyun_socket_status==ALIYUN_SERVER_DISCONNECTED)
                    goto reconnect;
                printf_rtos("Socket read timeout.Retrying..\n");
            }

            //发送部分
            //提取消息队列中的数据，发送到服务器

            MQTTString pubreq = MQTTString_initializer;
            pubreq.cstring=ALIYUN_MQTT_SERVER_PUB_TOPIC;//存放要发布的topic

            //JSON字符串接收指针
            char *json_str=NULL;
            //等待队列数据，没数据则略过
            if (xQueueReceive(sensor_send_queue,&json_str,0)==pdTRUE)
            {
                if (json_str!=NULL)
                {
                    //创建发布数据包
                    len=MQTTSerialize_publish(sendBuf,LENGTH_SENDBUF,0,0,0,1,pubreq,json_str,(int)strlen((char *)json_str));
                    //创建失败
                    if (len<=0)
                    {
                        printf_rtos("Can't serialize publish packet.\n");
                    }
                    else
                    {
                        int pubret=aliyun_transport_write(sendBuf,len);
                        //发送失败，重连
                        if (pubret<0)
                        {
                            printf_rtos("Publish failed, will reconnect.\n");
                            // The pointer is to a static buffer, so we must not free it.
                            goto reconnect;
                        }
                        else
                            printf_rtos("pub successfully.\n");
                    }
                    // The pointer is to a static buffer, so we must not free it.
                    json_str = NULL;
                }
            }

            //发送心跳包，当保活时间的一半到期时还没有有效活动，则发送心跳包
            //发送心跳包5S内如果没有接收到服务器的回应视为断开连接，重连
            if (xTaskGetTickCount() - last_active_tick>pdMS_TO_TICKS(ALIYUN_MQTT_SERVER_KEEP_ALIVE_INTERVAL_S*1000/2))
            {
                //首次发送心跳包
                if (!is_wait_pingresp)
                {
                    //配置发布topic
                    MQTTString pingreq = MQTTString_initializer;
                    pingreq.cstring=ALIYUN_MQTT_SERVER_PUB_TOPIC;
                    //生成，发送心跳包
                    len=MQTTSerialize_pingreq(sendBuf,LENGTH_SENDBUF);
                    ret=aliyun_transport_write(sendBuf,len);
                    //发送失败，重连
                    if (ret<0)
                        goto reconnect;
                    //发送成功，更新有效活动tick
                    last_pingreq_tick=xTaskGetTickCount();
                    //设置等待心跳回应标志位为真
                    is_wait_pingresp=true;
                    printf_rtos(">>> Send PINGREQ to server.\n");
                }
                else
                {
                    //发送心跳包5S内如果没有接收到服务器的回应视为断开连接，重连
                    if (xTaskGetTickCount()-last_pingreq_tick>pdMS_TO_TICKS(5000))
                    {
                        printf_rtos("--- PINGREQ timeout, connection lost --- \n");
                        //设置等待心跳回应标志位为假
                        is_wait_pingresp=false;
                        goto reconnect;
                    }
                }
            }
            size_t uxHighWaterMark = uxTaskGetStackHighWaterMark( NULL );
            printf_rtos("aliyunClient_task stack high water mark: %u words\n", uxHighWaterMark); // 假设你有log_debug
        }
        //重连逻辑，关闭socket，指数避退等等待
reconnect:
        closesocket(g_mqtt_aliyun_socket);
        g_mqtt_aliyun_socket=-1;
        printf_rtos("Reconnecting after %lu ms...\n", reconnect_delay * portTICK_PERIOD_MS);
        vTaskDelay(reconnect_delay);
        if (reconnect_delay < pdMS_TO_TICKS(60000))
            reconnect_delay *= 2; // 指数退避
    }


}
